package com.leo.biz.service.msgcenter;

import com.leo.biz.utils.ConfigUtils;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.message.MessageAndMetadata$;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by leo on 2016/11/16.
 */
@Service
public class MsgRecvService {

    static Logger logger = LoggerFactory.getLogger(MsgRecvService.class);
    private ExecutorService executor;
    private ConsumerConnector createConsumer() {
        Properties properties = new Properties();
        properties.put("zookeeper.connect", "192.168.32.128:2181");//声明zk
        properties.put("zookeeper.session.timeout.ms", "10000");
        properties.put("group.id", "0AAAAAAAA");// 必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据  同一组，则不能访问同一组内的topic数据
        return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
    }
    public  void start(String topic,final IRecvDataCallBack iRecvDataCallBack) {
        //String topic="mytopic";
        ConsumerConnector consumer = createConsumer();
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, 4); // 一次从主题中获取一个数据
        Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);
        //KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据
        List<KafkaStream<byte[], byte[]>> streamList = messageStreams.get(topic);// 获取每次接收到的这个数据>
        executor= Executors.newFixedThreadPool(4);
        int count=0;
        for(final KafkaStream<byte[], byte[]> streamTemp:streamList){
            final int threadNum=count;
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    logger.info("== Consume Thread Lanuched! threadNum="+threadNum+" ==");
                    ConsumerIterator<byte[], byte[]> iterator =  streamTemp.iterator();
                    while(iterator.hasNext()){
                        MessageAndMetadata messageAndMetadata=iterator.next();
                        iRecvDataCallBack.handle(messageAndMetadata);
                    }
                }
            });
            count++;
        }
    }

    public interface IRecvDataCallBack {
        public void handle(MessageAndMetadata data);
    }
}
